package com.crazymaker.demo.hystrix;

import lombok.extern.slf4j.Slf4j;
import org.junit.Test;
import rx.Observable;
import rx.functions.Func1;
import rx.functions.Func2;

import java.util.List;
import java.util.Random;
import java.util.concurrent.TimeUnit;

@Slf4j
public class WindowDemo {
    @Test
    public void hystrixTimewindowDemo() throws InterruptedException {
        // 创建Random
        Random random = new Random();

        // 事件流，100毫秒生成一次，1表示成功，0表示失败
        Observable<Integer> eventStream = Observable.interval(1000, TimeUnit.MILLISECONDS)
                .map(i -> random.nextInt(2));

        // 添加一个算子，300毫秒开窗统计一次，桶计数流
        Observable<Long> bucketedCounterStream = eventStream.window(3000, TimeUnit.MILLISECONDS)
                .flatMap(new Func1<Observable<Integer>, Observable<Long>>() {
                    @Override
                    public Observable<Long> call(Observable<Integer> eventBucket) {
                        Observable<List<Integer>> toList = eventBucket.toList();
                        return toList.map(list->{
                            long count = list.stream().filter(i -> i == 0).count();
                            log.info("{} '0 count:{}", list.toString(), count);
                            return count;
                        });
                    }
                });

        Observable<Object> bucketedRollingCounterStream = bucketedCounterStream.window(3, 1)
                .flatMap(new Func1<Observable<Long>, Observable<?>>() {
                    @Override
                    public Observable<?> call(Observable<Long> eventBucket) {
                        return eventBucket.reduce(new Func2<Long, Long, Long>() {
                            @Override
                            public Long call(Long aLong, Long aLong2) {
                                log.info("aLong = {}, aLong2 = {}", aLong, aLong2);
                                return aLong + aLong2;
                            }
                        });
                    }
                });
        bucketedRollingCounterStream.toBlocking().subscribe((l) -> {
            // toBlocking之后，当前逻辑在main线程执行
            log.info("滑动窗口的和：{}", l);
        });
        Thread.sleep(0x7fffffff);
    }

    @Test
    public void testRandom() {
        Random random = new Random();
        for (int i = 0; i < 10; i++) {
            log.info("random:{}", random.nextInt(2));
        }
    }
}
